package 电信数据清洗;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Map;

public class MapTest extends Mapper<LongWritable, Text, Data, NullWritable> {
    private Data k = new Data();
    private Map<String, String> userName = new HashMap<String, String>();
    private Map<String, String> location = new HashMap<String, String>();
    //用户姓名
    private String name_A;
    private String name_B;
    //用户地址
    private String loc_A;
    private String loc_B;
    //通话时间的转换
    private String startTime;
    private String endTime;
    //通话时间
    private String time;


    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        URI[] uris = context.getCacheFiles();
        File user = new File(uris[0]);
        String line;
        //缓存用户姓名信息
        BufferedReader br;
        br = new BufferedReader(new FileReader(user));
        while ((line = br.readLine()) != null) {
            userName.put(line.split(",")[1], line.split(",")[2]);
        }
        //缓存地址信息
        File loc = new File(uris[1]);
        br = new BufferedReader(new FileReader(loc));
        while ((line = br.readLine()) != null) {
            location.put(line.split(",")[1], line.split(",")[2]);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] data = value.toString().split(",");
        //将用户号码转换成姓名
        name_A = userName.get(data[0]);
        name_B = userName.get(data[1]);
        //将时间戳转换成日期类型
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        startTime = sdf.format(Long.parseLong(data[2]));
        endTime = sdf.format(Long.parseLong(data[3]));
        //计算通话时间
        time = Long.parseLong(data[3]) - Long.parseLong(data[2]) + "秒";
        //替换地址位置
        loc_A = location.get(data[4]);
        loc_B = location.get(data[5]);
        //写出数据
        k.set(name_A, name_B, data[0], data[1], startTime, endTime, time, loc_A, loc_B);
        context.write(k, NullWritable.get());
    }
}
